package udf.evaluation;

import java.io.IOException;

import java.io.UnsupportedEncodingException;
import java.net.URISyntaxException;

import java.util.LinkedList;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.data.DataType;

public class fetchGoogleSuggestions extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();
	String AGENT_LIST = "/home/sergio/projects/dictionary/user_agent_list.txt";

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	public LinkedList<String> getGoogleSuggestions(String query, String path,
			long delay) throws IOException, URISyntaxException {
		LinkedList<String> suggestions = null;

		suggestions = util.webSearch.GoogleQuery.getRelatedSearchesUDF(query,
				path, delay);

		return suggestions;
	}

	public DataBag exec(Tuple input) throws IOException {

		if (input.get(0) == null) {
			return null;
		}

		else {

			try {
				DataBag output = mBagFactory.newDefaultBag();

				String query = (String) input.get(0);
				String path = (String) input.get(1);
				Integer delay = (Integer) input.get(2);
				LinkedList<String> list = getGoogleSuggestions(query.trim(),
						path, delay);

				if (list == null) {
					TupleFactory mTupleFactory = TupleFactory.getInstance();
					Tuple tuple = mTupleFactory.newTuple();
					tuple.append(query);
					tuple.append("-error-");
					tuple.append(-1);
					//System.out.println(query + "\t" + "-error-" + "\t" + (-1));

				}else if (list.size()==0){
					TupleFactory mTupleFactory = TupleFactory.getInstance();
					Tuple tuple = mTupleFactory.newTuple();
					tuple.append(query);
					tuple.append("-no_suggestions-");
					tuple.append(-2);
					
				} else {

					for (int i = 0; i < list.size(); i++) {
						
						String expansion = list.get(i);
						TupleFactory mTupleFactory = TupleFactory.getInstance();
						Tuple tuple = mTupleFactory.newTuple();
						tuple.append(query);
						tuple.append(expansion);
						tuple.append(i + 1);
						//System.out.println(query + "\t" + expansion + "\t"	+ (i + 1));
						// Object tuple = mTupleFactory.newTuple(expansion);

						output.add(tuple);
					}
				}
				
				// StringTokenizer tok = new StringTokenizer((String)o,
				// " \",()*", false);
				// while (tok.hasMoreTokens())
				// output.add(mTupleFactory.newTuple(tok.nextToken()));
				// return output;

				return output;

			} catch (Exception e) {
				System.err.println("Failed to process input; error - "
						+ e.getMessage());
				return null;
			}

		}

	}

	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();

			bagSchema.add(new Schema.FieldSchema("query", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("suggestion",
					DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("rank", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}

}